package com.zz.common.dao

import java.time.Instant
import java.util.UUID

import scala.concurrent.duration._
import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success, Try}

import akka.actor.ActorSystem
import io.getquill._

import com.zz.ady.idl.IdType
import com.zz.common.cassandra.{CassandraConfig, CassandraExtension}
import com.zz.common.metrics.SimpleMetrics
import com.zz.idl.ZidIndex

object CassandraDAO {

  final case class Core(id: UUID, aid: Int, event: Array[Byte])

  final case class Event(id: UUID, aid: Int, event: Array[Byte])

  final case class Bc(id: UUID, aid: Int, t: Int, s: Int)

  def apply(cassandra: CassandraConfig)(implicit system: ActorSystem): CassandraDAO = {
    // val expire = ConfigFactory.load().getDuration("adx.metrics-expire").toMillis.milliseconds
    new CassandraDAO(cassandra)
  }

  def apply(system: ActorSystem): CassandraDAO = CassandraDAO(CassandraExtension(system).cassandra)(system)

  final case class Simple(id: UUID, t: Instant, vs: List[String])

  final case class T1(id: Int, t: Int, s: Int)

}

class CassandraDAO(cassandra: CassandraConfig, expire: Option[FiniteDuration] = None)(implicit system: ActorSystem) {

  import CassandraDAO._

  protected val logger = com.typesafe.scalalogging.Logger(getClass)

  private[this] val metrics = SimpleMetrics(system)

  // private[this] val expireInSeconds: Int = expire.toSeconds.toInt

  val ctx: CassandraAsyncContext[SnakeCase] = cassandra.ctxW

  import ctx._

  def f(x: Simple) = {
    val m = implicitly[InsertMeta[List[String]]]
    quote {
      query[Simple].insert(lift(x))
    }
  }

  def writeZidIndex(in: ZidIndex)(implicit ec: ExecutionContext): Future[Unit] = {
    val q = quote {
      query[ZidIndex].insert(lift(in)) //.usingTtl(lift(expireInSeconds))
    }
    val timer = metrics.scyllaDuration.labels("zid_index", "insert").startTimer()
    ctx
      .run(q)
      .andThen {
        case _ => timer.observeDuration()
      }
      .andThen {
        case Success(_) => metrics.scyllaCounter.labels("zid_index", "insert", "success", "1").inc()
        case Failure(_) => metrics.scyllaCounter.labels("zid_index", "insert", "failed", "0").inc()
      }
  }

  def findZidIndex(id: String)(implicit ec: ExecutionContext): Future[Option[ZidIndex]] = {
    val q = quote {
      query[ZidIndex].filter(_.id == lift(id)).take(1)
    }
    ctx.run(q).map(_.headOption)
  }

  def queryZidsByIdByType(id: String, `type`: IdType)(implicit ec: ExecutionContext): Future[List[String]] = {
    val q = quote {
      querySchema[(String, Int, List[String], Instant)](
          "reversed_index"
        , _._1 -> "id"
        , _._2 -> "type"
        , _._3 -> "zids"
        , _._4 -> "updated_at"
      ).filter(x => x._1 == lift(id) && x._2 == lift(`type`.value))
        .map(_._3)
        .take(1)
    }
    ctx.run(q).map(_.flatten)
  }

  implicit class RichListString(xs: Seq[String]) {
    def allEmpty: Boolean = xs.headOption.fold(false)(_.nonEmpty)
    def first: String     = xs.headOption.getOrElse("")
  }

  def queryZidsByIdBundle(bundle: ZidIndex)(implicit ec: ExecutionContext) = {

    val b = bundle

    val ids1: Vector[(List[String], IdType)] = Vector(
        (b.imei, IdType.IMEI)
      , (b.imeiMd5, IdType.IMEI_MD5)
      , (b.imeiSha1, IdType.IMEI_SHA1)
      , (b.imsi, IdType.IMSI)
      , (b.imsiMd5, IdType.IMSI_MD5)
      , (b.imsiSha1, IdType.IMSI_SHA1)
      , (b.androidId, IdType.ANDROID_ID)
      , (b.androidIdMd5, IdType.ANDROID_ID_MD5)
      , (b.androidIdSha1, IdType.ANDROID_ID_SHA1)
      , (b.mac, IdType.MAC)
      , (b.macMd5, IdType.MAC_MD5)
      , (b.macSha1, IdType.MAC_SHA1)
      , (b.sn, IdType.SN)
      , (b.idfa, IdType.IDFA)
      , (b.idfv, IdType.IDFV)
      , (b.mid, IdType.MID)
    )

    val ids2: Vector[(List[String], IdType)] = ids1.filter(_._1.nonEmpty).map {
      case (xs, t) =>
        xs.map(_.trim).filter(_.nonEmpty) -> t
    }

    val ids3: Vector[(String, Int)] = for {
      x       <- ids2
      (xs, t) = x
      y       <- xs
    } yield (y, t.value)

    val ids4 = ids3.map(_._1)

    val q = quote {
      querySchema[(String, Int, List[String], Long)](
          "reversed_index"
        , _._1 -> "id"
        , _._2 -> "type"
        , _._3 -> "zids"
        , _._4 -> "updated_at"
      ).filter { x =>
        liftQuery(ids4).contains(x._1)
      }
    }

    val timer = metrics.scyllaDuration.labels("zid_index", "query").startTimer()

    ctx
      .run(q)
      .map { xs =>
        logger.debug("rs: {}", xs)
        xs.filter(x => ids3.contains((x._1, x._2)))
          .flatMap(_._3)
      }
      .andThen {
        case Success(_) => metrics.scyllaCounter.labels("zid_index", "query", "success", "1").inc()
        case Failure(_) => metrics.scyllaCounter.labels("zid_index", "query", "failed", "0").inc()
      }
      .andThen {
        case _ => timer.observeDuration()
      }

  }

  def nonEmptyIds(b: ZidIndex): Vector[(List[String], IdType)] = {
    val ids1: Vector[(List[String], IdType)] = Vector(
        (b.imei, IdType.IMEI)
      , (b.imeiMd5, IdType.IMEI_MD5)
      , (b.imeiSha1, IdType.IMEI_SHA1)
      , (b.imsi, IdType.IMSI)
      , (b.imsiMd5, IdType.IMSI_MD5)
      , (b.imsiSha1, IdType.IMSI_SHA1)
      , (b.androidId, IdType.ANDROID_ID)
      , (b.androidIdMd5, IdType.ANDROID_ID_MD5)
      , (b.androidIdSha1, IdType.ANDROID_ID_SHA1)
      , (b.mac, IdType.MAC)
      , (b.macMd5, IdType.MAC_MD5)
      , (b.macSha1, IdType.MAC_SHA1)
      , (b.sn, IdType.SN)
      , (b.idfa, IdType.IDFA)
      , (b.idfv, IdType.IDFV)
      , (b.mid, IdType.MID)
    )

    val ids2: Vector[(List[String], IdType)] = ids1.filter(_._1.nonEmpty).map {
      case (xs, t) =>
        xs.map(_.trim).filter(_.nonEmpty) -> t
    }

    ids2
  }

  def putBundle(zid: String, b: ZidIndex)(implicit ec: ExecutionContext): Future[Unit] = {

    val t0 = System.currentTimeMillis()

    val ids2 = nonEmptyIds(b)

    val ids3: Vector[(String, IdType)] = for {
      x       <- ids2
      (xs, t) = x
      y       <- xs
    } yield (y, t)

    val reversed = ids3.map {
      case (id, t) =>
        (id, t.value, zid :: Nil, t0)
    }

    val z = b.copy(id = zid)

    val q1 = quote {
      query[ZidIndex].insert(lift(z))
    }

    val q2 = quote {
      querySchema[(String, Int, List[String], Long)](
          "reversed_index"
        , _._1 -> "id"
        , _._2 -> "type"
        , _._3 -> "zids"
        , _._4 -> "updated_at"
      )
    }

    val timer = metrics.scyllaDuration.labels("zid_index", "insert").startTimer()

    val f1: Future[Unit] =
      ctx
        .run(q1)
        .andThen {
          case Success(_) => metrics.scyllaCounter.labels("reversed_index", "query", "success", "1").inc()
          case Failure(_) => metrics.scyllaCounter.labels("reversed_index", "query", "failed", "0").inc()
        }
        .andThen {
          case _ => timer.observeDuration()
        }
    val f2: Future[Vector[Unit]] = Future.traverse(reversed) { x =>
      val timer2 = metrics.scyllaDuration.labels("reversed_index", "insert").startTimer()
      ctx
        .run(q2.insert(lift(x)))
        .andThen {
          case Success(_) => metrics.scyllaCounter.labels("zid_index", "insert", "success", "1").inc()
          case Failure(_) => metrics.scyllaCounter.labels("zid_index", "insert", "failed", "0").inc()
        }
        .andThen {
          case _ => timer2.observeDuration()
        }
    }
    val r1 = for {
      _ <- f1
      _ <- f2
    } yield ()
    r1.map(_ => zid)
  }

//  def f(xs: List[Int])(implicit ec: ExecutionContext) = {
//    val q = quote {
//      querySchema[(String, Int, List[UUID], Instant)](
//          "reversed_index"
//        , _._1 -> "id"
//        , _._2 -> "type"
//        , _._3 -> "zids"
//        , _._4 -> "updated_at"
//      ).filter(x => liftQuery(xs.toSet).contains(x._1))
//        .map(_._3)
//        .take(1)
//    }
//    ctx.run(q).map(_.flatten)
//  }

  def g(xs: List[Int])(implicit ec: ExecutionContext): Future[List[T1]] = {
    ctx.run(
        quote {
        query[T1].filter(x => liftQuery(xs.toSet) contains x.id)
      }
    )
  }

//  def writeCore(event: AdyCoreEvent)(implicit ec: ExecutionContext): Future[Unit] = {
//    {
//      for {
//        x <- Try(UUID.fromString(event.requestId))
//        y <- Try(event.adId)
//      } yield (x, y)
//    } match {
//      case Success((uuid, aid)) => this.writeCore(uuid, aid, event)
//      case Failure(exception) =>
//        exception.printStackTrace()
//        Future.successful(())
//    }
//  }

//  def writeCore(event: AdyEvent)(implicit ec: ExecutionContext): Future[Unit] =
//    this.writeCore(event.asCore)
//
//  def writeCore(uuid: UUID, aid: Int, event: AdyEvent)(implicit ec: ExecutionContext): Future[Unit] =
//    this.writeCore(uuid, aid, event.asCore)
//
//  def writeCore(uuid: UUID, aid: Int, event: AdyCoreEvent)(implicit ec: ExecutionContext): Future[Unit] = {
//    val t = Core(uuid, aid, event.toByteArray)
//    val q = quote {
//      query[Core].insert(lift(t)).usingTtl(lift(expireInSeconds))
//    }
//    val timer = SimpleMetrics.scyllaLatency.labels(cassandra.coreTable, "insert").startTimer()
//    ctx
//      .run(q)
//      .andThen {
//        case _ => timer.observeDuration()
//      }
//      .andThen {
//        case Success(_) => SimpleMetrics.scyllaCounter.labels(cassandra.coreTable, "insert", "success", "1")
//        case Failure(_) => SimpleMetrics.scyllaCounter.labels(cassandra.coreTable, "insert", "failed", "0")
//      }
//  }

//  def queryCoreEvent(id: String, aid: Int)(implicit ec: ExecutionContext): Future[List[AdyCoreEvent]] =
//    Try(UUID.fromString(id)).fold(_ => Future.successful(Nil), queryCoreEvent(_, aid))
//
//  def queryCoreEvent(id: UUID, aid: Int)(implicit ec: ExecutionContext): Future[List[AdyCoreEvent]] = {
//    val q = quote {
//      query[Core]
//        .filter(p => p.id == lift(id) && p.aid == lift(aid))
//        .map(_.event)
//    }
//    val timer = SimpleMetrics.scyllaLatency.labels(cassandra.coreTable, "query").startTimer()
//    ctx
//      .run(q)
//      .map(_.map(AdyCoreEvent.parseFrom))
//      .andThen {
//        case _ => timer.observeDuration()
//      }
//      .andThen {
//        case Success(x) if x.nonEmpty => SimpleMetrics.scyllaCounter.labels(cassandra.coreTable, "insert", "success", "1")
//        case Success(x)               => SimpleMetrics.scyllaCounter.labels(cassandra.coreTable, "insert", "success", "0")
//        case Failure(_)               => SimpleMetrics.scyllaCounter.labels(cassandra.coreTable, "insert", "failed", "0")
//      }
//  }

//  def queryBeacon(id: String, aid: Int)(implicit ec: ExecutionContext): Future[List[Bc]] =
//    Try(UUID.fromString(id)).fold(_ => Future.successful(Nil), queryBeacon(_, aid))
//
//  def queryBeacon(id: UUID, aid: Int)(implicit ec: ExecutionContext): Future[List[Bc]] = {
//    val q = quote {
//      query[Bc]
//        .filter(p => p.id == lift(id) && p.aid == lift(aid))
//    }
//    val timer = SimpleMetrics.scyllaLatency.labels(cassandra.beaconTable, "query").startTimer()
//    ctx
//      .run(q)
//      .andThen {
//        case _ => timer.observeDuration()
//      }
//      .andThen {
//        case Success(x) if x.nonEmpty => SimpleMetrics.scyllaCounter.labels(cassandra.beaconTable, "insert", "success", "1")
//        case Success(_)               => SimpleMetrics.scyllaCounter.labels(cassandra.beaconTable, "insert", "success", "0")
//        case Failure(_)               => SimpleMetrics.scyllaCounter.labels(cassandra.beaconTable, "insert", "failed", "0")
//      }
//  }

}
